package org.kafka.consumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class kafkaConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9093");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("group.id","consumer-test01");
        KafkaConsumer<String, Long> consumer = new KafkaConsumer<String, Long>(props);
        consumer.subscribe(Arrays.asList("test-out"));
        while (true) {
            ConsumerRecords<String, Long> records = consumer.poll(Duration.ofSeconds(10L));
            records.forEach(record->{
                System.out.println("key = [" + record.key() + "]"
                +"value = [" + record.value() + "]"
                +"time = [" + record.timestamp() + "]");
            });
        }
    }
}
